package com.iqik.rpc.boot;

import com.iqik.rpc.client.Client;
import com.iqik.rpc.service.UserService;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Watcher {
    private CuratorFramework client;
    private String path = "/server";
    private Map<String, Client> startServerMap = new HashMap<>();
    private RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

    public void init() {
        client = CuratorFrameworkFactory.builder()
                .connectString("132.232.16.135:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace("base") // 独立的命名空间 在/base目录下
                .build();
        client.start();
    }

    public void start() throws Exception {
        List<String> list = client.getChildren().forPath(path);
        System.out.println("现有运行的服务端：");
        list.forEach(string-> System.out.println(string));
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                try {
                    ChildData data = event.getData();
                    switch (event.getType()) {
                        case CHILD_UPDATED:
                        case CHILD_ADDED:
                            System.out.println(" add server " + data.getPath());
                            String addNodePath = data.getPath();
                            String addNode = addNodePath.substring(addNodePath.indexOf(path) + path.length() + 1);
                            String host = addNode.split(":")[0];
                            int port = Integer.parseInt(addNode.split(":")[1]);
                            Client addInitClient = new Client(host, port);
                            addInitClient.init();
                            if (addInitClient.getChannel().isSuccess()) {
                                startServerMap.put(addNode, addInitClient);
                            }
                            break;
                        case CHILD_REMOVED:
                            System.out.println(" delete server " + data.getPath());
                            String delNodePath = data.getPath();
                            String delNode = delNodePath.substring(delNodePath.indexOf(path) + path.length() + 1);
                            Client delInitClient = startServerMap.get(delNode);
                            delInitClient.stop();
                            if (!delInitClient.getChannel().isSuccess()) {
                                startServerMap.remove(delNode);
                            }
                            break;
                        default:
                            break;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    public void rpc(Client client) {
        UserService server = (UserService) client.createProxy(UserService.class, "UserService#sayHello#");
        String result = server.sayHello("Iqikia");
        System.out.println("result :" + result);
    }

    public static void main(String[] args) throws Exception {
        Watcher watcher = new Watcher();
        watcher.init();
        watcher.start();

        while (true) {
            for (Map.Entry<String, Client> entry : watcher.startServerMap.entrySet()) {
                String key = entry.getKey();
                System.out.println("address: " + key);
                Client client = watcher.startServerMap.get(key);//.sayHello("Iqikia");
                watcher.rpc(client);
            }
            Thread.sleep(2000);
        }
    }
}
